# -*- coding:utf-8 -*-
from __future__ import absolute_import, division, print_function, with_statement
from aiokafka import AIOKafkaProducer
import json


class Message(object):
    '''
    kafka消息订阅和生产，基于kafka-python==1.4.6
    有三个方法receive_message、register、producer_message
    其中register为装饰器，注册消息的handler
    * 'bootstrap_servers' kafka服务器,str或list (如：'115.29.76.68:9092' or ['115.29.76.68:9092'])
    * 'loop' asyncio.get_event_loop()
    '''

    def __init__(self, loop, bootstrap_servers):
        self.loop = loop
        self.bootstrap_servers = bootstrap_servers

    async def producer_message(self):
        '''
        * 'topic' str类型
        * 'value' dict类型,有具体的要求
        * 'key' uuid.uuid4().hex.encode() 可不指定
        '''
        pro = AIOKafkaProducer(
            loop=self.loop,
            bootstrap_servers=self.bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        return pro
